package com.shujia.gma.common

import org.apache.flink.configuration.Configuration
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.runtime.state.StateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.catalog.hive.HiveCatalog

abstract class FlinkTool {

  var bsEnv: StreamExecutionEnvironment = _
  var bsTableEnv: StreamTableEnvironment = _

  def main(args: Array[String]): Unit = {

    bsEnv = StreamExecutionEnvironment.getExecutionEnvironment


    /**
      * 开启checkpoint
      *
      */


    // 每 1000ms 开始一次 checkpoint
    bsEnv.enableCheckpointing(Config.getLong("checkpoint.interval"))

    // 高级选项：

    // 设置模式为精确一次 (这是默认值)
    bsEnv.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

    // 确认 checkpoints 之间的时间会进行 500 ms
    bsEnv.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)

    // Checkpoint 必须在一分钟内完成，否则就会被抛弃
    bsEnv.getCheckpointConfig.setCheckpointTimeout(60000)

    // 同一时间只允许一个 checkpoint 进行
    bsEnv.getCheckpointConfig.setMaxConcurrentCheckpoints(1)


    val config: CheckpointConfig = bsEnv.getCheckpointConfig
    //任务失败后自动保留最新的checkpoint文件
    config.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)

    //设置状态后端，保存状态的位置
    val stateBackend: StateBackend = new RocksDBStateBackend(Config.getString("checkpoint.path"), true)
    bsEnv.setStateBackend(stateBackend)


    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //使用blink的计划器
      .inStreamingMode() //使用流模型
      .build()

    //窗口table 环境
    bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)


    val configuration = new Configuration()
    //如果主键为null,自动删除
    configuration.setString("table.exec.sink.not-null-enforcer", "drop")
    configuration.setString("table.dynamic-table-options.enabled", "true")
    bsTableEnv.getConfig.addConfiguration(configuration)

    /**
      * 注册hive元数据
      *
      */
    val name = "myhive"
    val defaultDatabase = "flink"
    val hiveConfDir: String = Config.getString("hive.conf.dir")

    val hive = new HiveCatalog(name, defaultDatabase, hiveConfDir)
    //注册catalog
    bsTableEnv.registerCatalog("myhive", hive)

    // 切换catalog
    bsTableEnv.useCatalog("myhive")


    this.run(args)

  }

  def run(args: Array[String])

}
